package com.itheima.publisher;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@SpringBootTest
@Slf4j
public class publisher {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendMessage2Queue() {
        String queueName = "simple.queue";
        String msg = "hello,amqu!";
        rabbitTemplate.convertAndSend(queueName, msg);
    }

    @Test
    void testWorkQueue() throws InterruptedException {
        String queueName = "work.queue";
        for (int i = 0; i <=50 ; i++) {
            String msg = "message_"+i;
            rabbitTemplate.convertAndSend(queueName, msg);
            Thread.sleep(50);
        }
    }
    @Test
    void testSendFanout() {
        String exchangeName = "hmall.fanout";
        String msg = "hello,everyone!";
        rabbitTemplate.convertAndSend(exchangeName,null, msg);
    }
    @Test
    void testSendDirect() {
        String exchangeName = "hmall.direct";
        String msg = "hello,红色!";
        rabbitTemplate.convertAndSend(exchangeName,"yellow", msg);
    }
    @Test
    void testSendTopic() {
        String exchangeName = "hmall.topic";
        String msg = "hello,红色!";
        rabbitTemplate.convertAndSend(exchangeName,"china.news", msg);
    }
    @Test
    void testSend(){
        Map<String,Object>msg=new HashMap<>(2);
        msg.put("name","jack");
        msg.put("age",21);
        rabbitTemplate.convertAndSend("object.queue",msg);
    }
    @Test
    void testConfirmCallback() throws InterruptedException {
        //1.创建cd
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
        //2.添加ConfirmCallback
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息回调失败",ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                log.debug("收到confirm callback回执");
                if (result.isAck()){
                    //消息发送成功
                    log.debug("消息发送成功");
                }else {
                    //消息发送失败
                    log.error("消息发送失败，收到nack， 原因:{}",result.getReason());
                }
            }
        });
        rabbitTemplate.convertAndSend("hmall.direct","red","hello",cd);
        Thread.sleep(2000);
    }
    @Test
    void testSendTTLMessage(){
        Message message = MessageBuilder
                .withBody("hello".getBytes(StandardCharsets.UTF_8))
                .setExpiration("10000")
                .build();
        rabbitTemplate.convertAndSend("simple.direct","hi",message);
        log.info("消息发送成功");
    }
    @Test
    void testPublisherDelayMessage() {
        // 1.创建消息
        String message = "hello, delayed message";
        // 2.发送消息，利用消息后置处理器添加消息头
        rabbitTemplate.convertAndSend("delay.direct", "hi", message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                // 添加延迟消息属性
                message.getMessageProperties().setDelay(10000);
                return message;
            }
        });
    }
}
